package cn.mesmile.flink.window;

import cn.mesmile.flink.jdkstream.VideoOrder;
import cn.mesmile.flink.source.VideoOrderSourceV2;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author zb
 * @date 2021/8/23 22:46
 * @Description  CountWindows 统计 【滑动 / 滚动】
 */
public class FlinkCountWindowsApp {

    /**
     * 基于数量的滚动窗口, 滑动计数窗口
     * 案例：
     *      统计分组后 同个key内的数据超过5次 则进行统计   【滚动】 countWindow(5)
     *      分组后 组内数据超过2个 则触发统计过去的5个数据 【滑动】 countWindow(5, 2)
     * @param args
     */
    public static void main(String[] args) throws Exception {

//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        // 设置成 单个线程运行 方便查询演示效果
        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        DataStreamSource<VideoOrder> ds = env.addSource(new VideoOrderSourceV2());

        KeyedStream<VideoOrder, String> keyByDb = ds.keyBy(new KeySelector<VideoOrder, String>() {
            @Override
            public String getKey(VideoOrder videoOrder) throws Exception {
                return videoOrder.getTitle();
            }
        });
        // 分组后 组内数据超过 5 个则触发统计  【滚动】
//        DataStream<VideoOrder> money  = keyByDb.countWindow(5).sum("money");

        // 分组后 组内数据超过 2 个则触发统计过去的 5 个数据  【滑动】
        DataStream<VideoOrder> money = keyByDb.countWindow(5, 2).sum("money");
//        DataStream<VideoOrder> money = keyByDb.window(SlidingProcessingTimeWindows.of(Time.seconds(10),Time.seconds(5))).sum("money");

        money.print("result:");

        env.execute("count windows job");
    }
}
